/**
 * 
 */
package com.pxingg.model.pubsub;

import junit.framework.Assert;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.json.JSONException;
import org.json.JSONObject;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import com.pxingg.Global;
import com.pxingg.datetime.TimeUtils;
import com.pxingg.datetime.Timer;
import com.pxingg.log.LogConfig;

/**
 * @author pxingg
 * 
 */
public class MessageTest
{
    /**
     * 模拟一个订阅者对消息进行测试
     * 
     * @author pxingg
     * 
     */
    protected class TSubscriber extends Subscriber
    {
        int counter = 0;

        /**
         * @param name
         */
        public TSubscriber(final String name)
        {
            super(name);
        }

        public int getCounter()
        {
            return counter;
        }

        /**
         * 使用计数器+1的方式表示这个成功执行了一个消息订阅
         */
        @Override
        public void process(final String subscriberMessageName, final Message message)
        {
            super.process(subscriberMessageName, message);

            synchronized (Subscriber.log)
            {
                counter++;
            }

            // 标记消息执行成功
            message.setCurrentResultSuccess();
        }
    }

    /**
     * The log of this class
     */
    static Log log = LogFactory.getLog(MessageTest.class);

    /**
     * 测试的消息发布者
     */
    static String messageTest = "testMessage";

    /**
     * 测试消息的发布点
     */
    static String publishPoint = "com.pxingg.messageTest";

    /**
     * 测试消息的所有者
     */
    static String messageOwner = "MessageTest.class";

    /**
     * 生成一个测试的消息对象
     * 
     * @return 一个测试的消息对象
     */
    static Message getNewMessage()
    {
        return new Message(MessageTest.messageTest, MessageTest.messageOwner, MessageTest.publishPoint);
    }

    /**
     * @throws java.lang.Exception
     */
    @BeforeClass
    public static void setUpBeforeClass() throws Exception
    {
        // 打开日志开关
        LogConfig.init();
    }

    /**
     * @throws java.lang.Exception
     */
    @AfterClass
    public static void tearDownAfterClass() throws Exception
    {
        // Global.switchCenter.shutdown();
    }

    /**
     * @throws java.lang.Exception
     */
    @Before
    public void setUp() throws Exception
    {
    }

    /**
     * @throws java.lang.Exception
     */
    @After
    public void tearDown() throws Exception
    {
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#containsKey(java.lang.String)}.
     */
    @Test
    public final void testContainsKey()
    {
        final Message message = new Message(MessageTest.messageTest);
        message.setData("key1", "value1");
        Assert.assertTrue(message.containsKey("key1"));
        Assert.assertFalse(message.containsKey("notExistKey"));
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#getMessageID()}.
     */
    @Test
    public final void testGetMessageID()
    {
        final Message message = new Message(MessageTest.messageTest);
        // 在一开始，消息ID从0000000001开始，所以最开始的肯定一000000开始
        Assert.assertTrue(message.getMessageID().startsWith("0000000"));
        // 下一个消息的ID肯定比上一个大
        Assert.assertTrue(new Message(MessageTest.messageTest).getMessageID().compareTo(message.getMessageID()) > 0);
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#getMessagePulisher()}.
     */
    @Test
    public final void testGetMessagePulisher()
    {
        final Message message = new Message(MessageTest.messageTest);

        Assert.assertEquals(MessageTest.messageTest, message.getMessageName());
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#getNewResultContainer(String, String)}<br>
     * {@link com.pxingg.model.pubsub.Message#getCurrentResultContainer()}
     */
    @Test
    public final void testGetNewResultContainer()
    {
        final Message message = new Message(MessageTest.publishPoint);
        JSONObject json = message.getNewResultContainer("subscriberPoint", "subscriber");

        Assert.assertEquals("subscriberPoint", json.optString("subscriberPoint"));
        Assert.assertEquals("subscriber", json.optString("subscriber"));

        // 一个订阅者生成了执行结果
        Assert.assertEquals(1, message.getSubscriberCounter());

        JSONObject currentJson = message.getCurrentResultContainer();
        Assert.assertEquals("subscriberPoint", currentJson.optString("subscriberPoint"));
        Assert.assertEquals("subscriber", currentJson.optString("subscriber"));

        // 第二次调用，生成一个新的结果集
        json = message.getNewResultContainer("subscriberPoint", "subscriber");

        // 两个订阅者生成了执行结果
        Assert.assertEquals(2, message.getSubscriberCounter());

        currentJson = message.getCurrentResultContainer();
        Assert.assertEquals("subscriberPoint", currentJson.optString("subscriberPoint"));
        Assert.assertEquals("subscriber", currentJson.optString("subscriber"));
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#isFinishPublish()}.
     */
    @Test
    public final void testIsFinishPublish()
    {
        Global.switchCenter.removeAllSubscriber();
        final Message message = new Message(MessageTest.messageTest);
        message.publish();

        final Timer timer = new Timer();
        timer.start();

        while (timer.getInterval() < 50)
        {
            if (message.isFinishPublish())
            {
                break;
            }
        }

        Assert.assertTrue(message.isFinishPublish());
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#Message(java.lang.String, java.lang.String, java.lang.String)}.<br>
     * Test method for {@link com.pxingg.model.pubsub.Message#Message(java.lang.String, java.lang.String)}.<br>
     * Test method for {@link com.pxingg.model.pubsub.Message#Message(java.lang.String)}.
     */
    @Test
    public final void testMessage()
    {
        Message message = new Message(MessageTest.messageTest, MessageTest.messageOwner, MessageTest.publishPoint);
        Assert.assertEquals(MessageTest.messageTest, message.getMessageName());
        Assert.assertEquals(MessageTest.messageOwner, message.getMessageOwner());
        Assert.assertEquals(MessageTest.publishPoint, message.getPublishPoint());

        // 在一开始，消息ID从0000000001开始，所以最开始的肯定一000000开始
        Assert.assertTrue(message.getMessageID().startsWith("0000000"));
        // 下一个消息的ID肯定比上一个大
        Assert.assertTrue(new Message(MessageTest.messageTest).getMessageID().compareTo(message.getMessageID()) > 0);

        message = new Message(MessageTest.messageTest, MessageTest.messageOwner);
        // 这时消息名等于发布点
        Assert.assertEquals(MessageTest.messageTest, message.getPublishPoint());

        message = new Message(MessageTest.messageTest);
        // 这时消息名等于发布点
        Assert.assertEquals(Message.MESSAGE_OWNER_UNDEFINED, message.getMessageOwner());
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#publish()}. 发布出去消息后，switchCenter会自动处理该消息
     */
    @Test
    public final void testPublish()
    {
        Global.switchCenter.removeAllSubscriber();
        final Message message = new Message(MessageTest.publishPoint);
        message.setData("key1", "value1");
        final TSubscriber subscriber = new TSubscriber(MessageTest.messageTest);
        subscriber.subscribe(MessageTest.publishPoint);
        int oldCounter = subscriber.getCounter();
        // 消息发布了三次
        message.publish();
        new Message(MessageTest.publishPoint).publish();
        new Message(MessageTest.publishPoint).publish();
        TimeUtils.mySleep(50);
        Assert.assertEquals(3, subscriber.getCounter() - oldCounter);

        // 消息从跟节点上被订阅了两次
        oldCounter = subscriber.getCounter();
        subscriber.subscribe("com.pxingg");
        subscriber.subscribe("com.notexist"); // 这个订阅没有消除需要处理
        new Message(MessageTest.publishPoint).publish();
        TimeUtils.mySleep(50);
        Assert.assertEquals(2, subscriber.getCounter() - oldCounter);
        subscriber.unsubscribeAll();
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#publishAtOnce()}. 发布出去消息后，当前线程会自动调用switchCenter处理该消息
     */
    @Test
    public final void testPublishAtOnce()
    {
        Global.switchCenter.removeAllSubscriber();
        final Message message = new Message(MessageTest.publishPoint);
        message.setData("key1", "value1");
        final TSubscriber subscriber = new TSubscriber(MessageTest.messageTest);
        subscriber.subscribe(MessageTest.publishPoint);
        int oldCounter = subscriber.getCounter();
        // 消息发布了三次
        message.publishAtOnce();
        new Message(MessageTest.publishPoint).publishAtOnce();
        new Message(MessageTest.publishPoint).publishAtOnce();
        Assert.assertEquals(3, subscriber.getCounter() - oldCounter);

        // 消息从跟节点上被订阅了两次
        oldCounter = subscriber.getCounter();
        subscriber.subscribe("com.pxingg");
        subscriber.subscribe("com.notexist"); // 这个订阅没有消除需要处理
        new Message(MessageTest.publishPoint).publishAtOnce();
        Assert.assertEquals(2, subscriber.getCounter() - oldCounter);
        subscriber.unsubscribeAll();
    }

    /**
     * Test method for {@link com.pxingg.message.Message#setCurrentResultFailure()()}<br>
     */
    @Test
    public final void testSetCurrentResultFailure()
    {
        final Message message = new Message(MessageTest.publishPoint);
        message.getNewResultContainer("subscriberPoint", "subscriber");

        message.setCurrentResultFailure();

        Assert.assertEquals(Message.RESULT_FAILURE, message.getCurrentResultContainer().optString("result"));
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#setCurrentResultSuccess()}<br>
     */
    @Test
    public final void testSetCurrentResultSuccess()
    {
        final Message message = new Message(MessageTest.publishPoint);
        message.getNewResultContainer("subscriberPoint", "subscriber");

        message.setCurrentResultSuccess();

        Assert.assertEquals(Message.RESULT_SUCCESS, message.getCurrentResultContainer().optString("result"));
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#setData(java.lang.String, java.lang.String)}.<br>
     * Test method for {@link com.pxingg.model.pubsub.Message#getData(java.lang.String)}.
     */
    @Test
    public final void testSetDataGetData()
    {
        final Message message = new Message(MessageTest.messageTest);
        message.setData("key1", "value1").setData("key2", "value2").setData("key3", "value3");
        Assert.assertEquals("value1", message.getData("key1"));
        Assert.assertEquals("value2", message.getData("key2"));
        Assert.assertEquals("value3", message.getData("key3"));

        // 新的值会覆盖老的值
        message.setData("key1", "newvalue1");
        Assert.assertEquals("newvalue1", message.getData("key1"));

        // 如果key不存在，那么获取的值为空
        Assert.assertEquals("", message.getData("notExistKey"));

        // 或者这个消息的JSON的对象
        final JSONObject json = message.getData();
        Assert.assertEquals("newvalue1", json.optString("key1"));
        Assert.assertEquals("value2", json.optString("key2"));
        Assert.assertEquals("value3", json.optString("key3"));
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#setResult()}.<br>
     * Test method for {@link com.pxingg.model.pubsub.Message#getResult()}. Test method for {@link com.pxingg.model.pubsub.Message#getSubscriberCounter()}.
     */
    @Test
    public final void testSetResult()
    {
        Global.switchCenter.removeAllSubscriber();
        final TSubscriber subscriber = new TSubscriber(MessageTest.messageTest);
        subscriber.subscribe(MessageTest.publishPoint);

        final Message message = MessageTest.getNewMessage();
        message.publish();

        final Timer timer = new Timer();
        timer.start();

        while (timer.getInterval() < 50)
        {
            if (message.isFinishPublish())
            {
                break;
            }
        }
        Assert.assertTrue(message.isFinishPublish());

        Assert.assertEquals(1, message.getSubscriberCounter());

        // 只有一个订阅，并且订阅的结果显示执行成功
        try
        {
            Assert.assertEquals(Message.RESULT_SUCCESS, message.getResult().get(0).get("result"));
        }
        catch (final JSONException e)
        {
            Assert.fail("Assert.assertEquals(Message.RESULT_SUCCESS, message.getResult().get(0).get(\"result\"))");
        }
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#isTimeOut()}.<br>
     * {@link com.pxingg.model.pubsub.Message#setTimeOut()}<br>
     * {@link com.pxingg.model.pubsub.Message#getTimeOut()}<br>
     */
    @Test
    public final void testTimeOut()
    {
        Global.switchCenter.removeAllSubscriber();
        final Message message = new Message(MessageTest.publishPoint);
        message.setData("key", "value");
        final TSubscriber subscriber = new TSubscriber("TSubscriber");
        subscriber.subscribe(MessageTest.publishPoint);

        message.setTimeout(1);

        TimeUtils.mySleep(10);
        // 消息发布了三次
        message.publishAtOnce();
        Assert.assertEquals(1, message.getSubscriberCounter());

        // 消息超时，不处理
        final JSONObject json = message.getResult().get(0);
        Assert.assertEquals(Message.RESULT_TIMEOUT, json.optString("result"));

        subscriber.unsubscribeAll();
    }

    /**
     * Test method for {@link com.pxingg.model.pubsub.Message#toString()}.
     */
    @Test
    public final void testToString()
    {
        final Message message = new Message(MessageTest.messageTest);
        message.setData("key1", "value1");

        Assert.assertTrue(message.toString().contains("\"key1\":\"value1\""));
    }
}
